package com.offcn.bigdata.spark.p3

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
  * 分组排序:
  *     因为这里使用groupByKey，所以效率不是很高，需要使用combineByKey来去代替它
  */
object _11GroupSortOps {
    def main(args: Array[String]): Unit = {
        val conf = new SparkConf()
            .setMaster("local[*]")
            .setAppName(s"${_11GroupSortOps.getClass.getSimpleName}")
        val sc = new SparkContext(conf)

        val lines = sc.textFile("file:/E:/work/2020-0828期大数据/workspace/spark-parent-0828/data/topn.txt")

        val course2Info:RDD[(String, String)] = lines.map(line => {
            val course = line.substring(0, line.indexOf(" "))
            val info = line.substring(line.indexOf(" ") + 1)
            (course, info)
        })
        //分组
        val course2Infos:RDD[(String, Array[String])] = course2Info.combineByKey(
            createCombiner,
            mergeValue,
            mergeCombiners
        )
        println("-----------------------分组之后的内容-----------------------------")
        course2Infos.foreach{case (course, infos) => {
            println(s"course: ${course}, info: ${infos.mkString("[", ", ", "]")}")
        }}
        sc.stop()
    }

    def createCombiner(info: String): Array[String] = Array(info)
    def mergeValue(infos: Array[String], info:String): Array[String] = {
        infos.+:(info).sorted(new Ordering[String](){
            override def compare(x: String, y: String): Int = {
                val scoreX = x.split("\\s+")(1).toDouble
                val scoreY = y.split("\\s+")(1).toDouble
                scoreY.compareTo(scoreX)
            }
        }).take(3)
    }
    def mergeCombiners(infos1: Array[String], infos2: Array[String]): Array[String] = {
        val ret = infos1 ++ infos2

        ret.sorted(new Ordering[String](){
            override def compare(x: String, y: String): Int = {
                val scoreX = x.split("\\s+")(1).toDouble
                val scoreY = y.split("\\s+")(1).toDouble
                scoreY.compareTo(scoreX)
            }
        }).take(3)
    }
}
